package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
 *
 * 当窗口中有10个元素时计算一次Wordcount
 * */

object FlinkTest_17_window_count {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      .countWindow(10, 10)
      .reduce(new ReduceFunction[(String, Int)] {
        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {
          (value1._1, value1._2 + value2._2)
        }
      }).print()


    env.execute()
  }

}
